package com.yifeng.repo.flink.data.transport.bootstrap;

import com.yifeng.repo.flink.data.transport.config.JobParametersConfig;
import com.yifeng.repo.flink.data.transport.config.SinkIcebergConfig;
import com.yifeng.repo.flink.data.transport.dto.IcebergTableInfo;
import com.yifeng.repo.flink.data.transport.streaming.iceberg.FlinkSinkIcebergBuilder;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.iceberg.actions.RewriteDataFilesActionResult;
import org.apache.iceberg.flink.actions.Actions;

import java.util.List;

/**
 * 合并iceberg的 data files 小文件
 * @author wangzhi
 * @since 2023-06-13
 */
public class RewriteIcebergDataFiles {
    public static void main(String[] args) throws Exception {
        //获取配置
        JobParametersConfig config = BootstrapHelper.setEnvAndParameter(args);
        SinkIcebergConfig sinkIcebergConfig = config.getSinkIcebergConfig();
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //配置iceberg 库名和表名并加载表
        List<IcebergTableInfo> icebergTables = FlinkSinkIcebergBuilder.buildIcebergTableInfo(sinkIcebergConfig);
        for(IcebergTableInfo tableInfo : icebergTables) {

            //合并 data files 小文件
            RewriteDataFilesActionResult result = Actions.forTable(tableInfo.getTable())
                    .rewriteDataFiles()
                    //默认 512M ，可以手动通过以下指定合并文件大小，与Spark中一样。
                    .targetSizeInBytes(536870912L)
                    .execute();
        }
    }
}
